package sf.r2dbc.sql;

import io.r2dbc.spi.Connection;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import sf.common.wrapper.Page;
import sf.core.DBObject;
import sf.database.dao.DBContext;
import sf.database.dialect.DBDialect;
import sf.database.jdbc.sql.CrudExampleImpl;
import sf.database.jdbc.sql.SQLContext;
import sf.database.meta.TableMapping;
import sf.database.support.DMLType;
import sf.database.util.OrmValueUtils;
import sf.dsl.DeleteDSL;
import sf.dsl.InsertDSL;
import sf.dsl.UpdateDSL;
import sf.dsl.example.Example;
import sf.dsl.example.SelectOpt;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

/**
 * 按example操作
 */
public class R2dbcCrudExampleImpl implements R2dbcCrudExampleInf {
    private static R2dbcCrudExampleImpl instance = new R2dbcCrudExampleImpl();

    public static R2dbcCrudExampleImpl getInstance() {
        return instance;
    }

    private R2dbcCrudExampleImpl() {

    }

    @Override
    public <T> Mono<Page<T>> selectPage(Connection conn, long start, int limit, Class<T> beanClass, Example example) {
        DBDialect dialect = R2dbcUtils.getDBDialect(conn);
        DBContext dbContext = R2dbcUtils.getDBContext(conn);
        SQLContext select = getQuerySelect(dbContext, dialect, example, false, example.isCount() ? SelectOpt.count : null, false);
        return R2dbcCrud.getInstance().getCrudModel().selectPage(conn, start, limit, beanClass, select);
    }

    @Override
    public Mono<Object[]> selectArray(Connection conn, Example example) {
        return selectOne(conn, Object[].class, example);
    }

    @Override
    public <T> Flux<T> selectList(Connection conn, Class<T> beanClass, Example example) {
        DBDialect dialect = R2dbcUtils.getDBDialect(conn);
        DBContext dbContext = R2dbcUtils.getDBContext(conn);
        SQLContext select = getQuerySelect(dbContext, dialect, example, false, example.isCount() ? SelectOpt.count : null);
        return R2dbcCrud.getInstance().getCrudModel().selectList(conn, beanClass, select);
    }

    @Override
    public <T> Flux<T> selectList(Connection conn, Class<T> beanClass, long start, int limit, Example example) {
        DBDialect dialect = R2dbcUtils.getDBDialect(conn);
        DBContext dbContext = R2dbcUtils.getDBContext(conn);
        SQLContext select = getQuerySelect(dbContext, dialect, example, false, example.isCount() ? SelectOpt.count : null, false);
        return R2dbcCrud.getInstance().getCrudModel().selectList(conn, beanClass, select, start, limit);
    }

    @Override
    public <T> Flux<T> selectListForUpdate(Connection conn, Class<T> beanClass, Example example) {
        DBDialect dialect = R2dbcUtils.getDBDialect(conn);
        DBContext dbContext = R2dbcUtils.getDBContext(conn);
        SQLContext select = getQuerySelect(dbContext, dialect, example, true, example.isCount() ? SelectOpt.count : null);
        return R2dbcCrud.getInstance().getCrudModel().selectList(conn, beanClass, select);
    }

    @Override
    public <T> Mono<T> selectOne(Connection conn, Class<T> beanClass, Example example) {
        DBDialect dialect = R2dbcUtils.getDBDialect(conn);
        DBContext dbContext = R2dbcUtils.getDBContext(conn);
        SQLContext select = getQuerySelect(dbContext, dialect, example, false, example.isCount() ? SelectOpt.count : null);
        return R2dbcCrud.getInstance().getCrudModel().selectOne(conn, beanClass, select);
    }

    @Override
    public <T> Mono<T> selectOneForUpdate(Connection conn, Class<T> beanClass, Example example) {
        DBDialect dialect = R2dbcUtils.getDBDialect(conn);
        DBContext dbContext = R2dbcUtils.getDBContext(conn);
        SQLContext select = getQuerySelect(dbContext, dialect, example, true, example.isCount() ? SelectOpt.count : null);
        return R2dbcCrud.getInstance().getCrudModel().selectOne(conn, beanClass, select);
    }

    @Override
    public Flux<Map<String, Object>> select(Connection conn, Example example) {
        DBDialect dialect = R2dbcUtils.getDBDialect(conn);
        DBContext dbContext = R2dbcUtils.getDBContext(conn);
        SQLContext select = getQuerySelect(dbContext, dialect, example, false, example.isCount() ? SelectOpt.count : null);
        return R2dbcCrud.getInstance().getCrudModel().selectListMap(conn, select);
    }

    /**
     * @param context
     * @param dialect
     * @param query
     * @param forUpdate
     * @param opt
     * @return
     */
    private static SQLContext getQuerySelect(DBContext context, DBDialect dialect, Example query, boolean forUpdate, SelectOpt opt) {
        return getQuerySelect(context, dialect, query, forUpdate, opt, true);//默认使用example中的分页参数
    }

    /**
     * @param context
     * @param dialect
     * @param example     查询条件
     * @param forUpdate   是否需要加入 for update
     * @param opt         操作
     * @param offsetLimit 是否使用Example中的offsetLimit
     * @return
     */
    private static SQLContext getQuerySelect(DBContext context, DBDialect dialect, Example example, boolean forUpdate, SelectOpt opt, boolean offsetLimit) {
        return CrudExampleImpl.getQuerySelect(context, dialect, example, forUpdate, opt, offsetLimit);//默认使用example中的分页参数
    }

    @Override
    public Mono<Integer> insert(Connection conn, InsertDSL insertDSL, List<DBObject> datas) {
        DBDialect dialect = R2dbcUtils.getDBDialect(conn);
        insertDSL.setDialect(dialect);
        List<SQLContext> sqlContext = insertDSL.getSQLContexts();
        TableMapping tm = insertDSL.getEntity().getTableMapping();
        if (datas != null) {
            int count = insertDSL.getCount();
            for (int i = 0; i < count; i++) {
                DBObject object = OrmValueUtils.instance(tm);
                datas.add(object);
            }
        } else {
            datas = Collections.emptyList();
        }
        return insertBatch(conn, insertDSL.isBatchToBulk(), sqlContext, datas, tm, dialect);
    }

    private Mono<Integer> insertBatch(Connection conn, boolean batchToBulk, List<SQLContext> list, List<DBObject> datas, TableMapping table, DBDialect dialect) {
        Map<String, List<SQLContext>> map = new HashMap<>();
        for (SQLContext b : list) {
            String sql = b.getSql();
            List<SQLContext> objects = map.computeIfAbsent(sql, k -> new ArrayList<>());
            objects.add(b);
        }
        Flux<Integer> flux = Flux.empty();
        if (!map.isEmpty()) {
            boolean first = true;
            int j = 0;
            for (Map.Entry<String, List<SQLContext>> entry : map.entrySet()) {
                String sql = entry.getKey();
                List<SQLContext> sqlContextList = entry.getValue();
                List<Integer> dataSizeList = new LinkedList<>();
                List<DBObject> dataList = new LinkedList<>();
                if (batchToBulk) {
                    dataSizeList.add(datas.size());
                    dataList = datas;
                } else {
                    for (int i = 0; i < sqlContextList.size(); i++) {
                        dataSizeList.add(1);
                        if (j < datas.size()) {
                            dataList.add(datas.get(j++));
                        }
                    }
                }
                Flux<Integer> tempFlux = R2dbcCrud.getInstance().getCrudModel().batch(conn, dialect, table, sql, sqlContextList, dataSizeList, dataList, DMLType.INSERT, false);
                if (first) {
                    flux = tempFlux;
                } else {
                    flux = flux.mergeWith(tempFlux);
                }
                first = false;
            }
        }
        return flux.count().cast(Integer.class);
    }

    @Override
    public Mono<Integer> update(Connection conn, UpdateDSL updateDSL) {
        DBDialect dialect = R2dbcUtils.getDBDialect(conn);
        updateDSL.setDialect(dialect);
        SQLContext sqlContext = updateDSL.getSQLContext();
        return R2dbcCrud.getInstance().getCrudModel().execute(conn, sqlContext);
    }

    @Override
    public Mono<Integer> delete(Connection conn, DeleteDSL deleteDSL) {
        DBDialect dialect = R2dbcUtils.getDBDialect(conn);
        deleteDSL.setDialect(dialect);
        SQLContext sqlContext = deleteDSL.getSQLContext();
        return R2dbcCrud.getInstance().getCrudModel().execute(conn, sqlContext);
    }
}
